package cn.like.stream.core;

import cn.like.function.Func;
import cn.like.stream.helper.MessageSplitter;
import cn.like.stream.helper.RocketMessageCreator;
import org.apache.poi.ss.formula.functions.T;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.MessagingException;

import java.util.List;
import java.util.function.Supplier;

/**
 * Description: 消息发送者 <br>
 * <pre>
 *     tips:
 *      1.消息推荐使用{@link cn.like.stream.helper.RocketMessageCreator} 创建
 * </pre>
 *
 * @author like 980650920@qq.com
 * @date 2021-07-18 10:14:06
 */
public interface Sender extends Rocket {

    Logger log = LoggerFactory.getLogger(Sender.class);

    //  SYNC

    /**
     * 发送消息
     *
     * @param message 消息
     * @return {@link SendResult}
     */
    default SendResult send(Message message) {
        try {
            long now = System.currentTimeMillis();
            if (Func.isBlack.test(message.getTopic())) {
                message.setTopic(topic());
            }
            SendResult sendResult = rocketMQTemplate().getProducer().send(message, SEND_TIME_OUT);
            long costTime = System.currentTimeMillis() - now;
            if (log.isDebugEnabled()) {
                log.debug("send message cost: {} ms, msgId:{}", costTime, sendResult.getMsgId());
            }
            return sendResult;
        } catch (Exception e) {
            log.error("syncSend failed. destination:{}, message:{} ", message.getTopic(), message);
            throw new MessagingException(e.getMessage(), e);
        }
    }

    /**
     * 发送批量消息
     * <pre>
     *     1.message 中 必须设置topic
     *     2.message 不能设置 delay level
     * </pre>
     *
     * @param messages 消息
     */
    default void sendBatch(List<Message> messages) {
        //把大的消息分裂成若干个小的消息
        MessageSplitter splitter = new MessageSplitter(messages);
        while (splitter.hasNext()) {
            List<org.apache.rocketmq.common.message.Message> listItem = splitter.next();
            try {
                rocketMQTemplate().getProducer().send(listItem, SEND_TIME_OUT);
            } catch (Exception e) {
                e.printStackTrace();
                //处理error
                handlerSendBatchError(listItem);
            }
        }
    }

    default void sendBatch(Supplier<List<Message>> supplier) {
        sendBatch(buildMessageList(supplier));
    }

    default void handlerSendBatchError(List<org.apache.rocketmq.common.message.Message> listItem) {
        log.error("[handlerSendBatchError] -  消息发送失败,数量： {} - topic:{}", listItem.size(), listItem.get(0).getTopic());
    }

    // async

    /**
     * 发送异步消息
     *
     * @param message      消息
     * @param sendCallback 发送回调
     */
    default void sendAsync(Message message, SendCallback sendCallback) {
        try {
            if (Func.isBlack.test(message.getTopic())) {
                message.setTopic(topic());
            }
            rocketMQTemplate().getProducer().send(message, sendCallback, SEND_TIME_OUT);
        } catch (Exception e) {
            log.error("syncSend failed. destination:{}, message:{} ", message.getTopic(), message);
            throw new MessagingException(e.getMessage(), e);
        }
    }

    // tool
    default Message buildMessage(T body) {
        return RocketMessageCreator.of(topic(),body).create();
    }

    default List<Message> buildMessageList(Supplier<List<Message>> supplier) {
        return supplier.get();
    }

}

